/// @brief 发布订阅模式
#ifndef __UTOOLS_PUBSUB_H__
#define __UTOOLS_PUBSUB_H__

#include "../utools_cfg.h"

#if UTOOLS_PUBSUB_ENABLE
#include <unordered_map>
#include <vector>
#include <functional>
#include <memory>
#include <type_traits>
#include <cstdint>
#include <cstddef>
#include <algorithm>

#if (UTOOLS_COLLAB_TASK_ENABLE && UTOOLS_COLLAB_SYNC_RANDOM_ACCESS_QUEUE_ENABLE)
#include "../collab/utmutex.h"
#include "../collab/collab_task.h"
#include "../collab/sync_random_access_queue.h"
#else
#error "PubSub requires UTOOLS_COLLAB_TASK_ENABLE and UTOOLS_COLLAB_SYNC_RANDOM_ACCESS_QUEUE_ENABLE."
#endif // UTOOLS_COLLAB_TASK_ENABLE
#endif // UTOOLS_EVENT_BUS_ENABLE

#if UTOOLS_SYNC_EVENT_BUS_ENABLE

#endif // UTOOLS_EVENT_BUS_ENABLE

namespace utools::pattern
{
#if (UTOOLS_PUBSUB_ENABLE && UTOOLS_COLLAB_TASK_ENABLE)
    /// @brief 发布订阅模式
    /// @tparam TOPIC_TYPE 消息主题类型
    /// @tparam BUFFER_SIZE 缓存的长度，如果为0，会自动设置为1024
    template <typename TOPIC_TYPE, std::size_t BUFFER_SIZE = 1024>
    class PubSub
    {
    public:
        typedef TOPIC_TYPE topic_type;
        typedef std::shared_ptr<void> message_type;

        /// @brief 消息包装器，用于提供给回调函数使用，方便数据解析
        typedef struct message_wrapper_type
        {
        private:
            message_type ptr{nullptr}; // 消息内容

        public:
            message_wrapper_type(const message_type message) : ptr(message) {}

            /// @brief 获取消息内容
            /// @tparam T 需要要解析的消息类型
            /// @return 消息内容
            /// @note 建议使用共享指针cast_ptr()返回，避免大对象拷贝
            template <typename T>
            const T cast() const
            {
                return *std::static_pointer_cast<T>(ptr);
            }

            /// @brief 获取消息内容
            /// @tparam T 需要要解析的消息类型
            /// @return 消息内容
            /// @note 建议使用共享指针cast_ptr()返回，避免大对象拷贝
            template <typename T>
            T cast()
            {
                return *std::static_pointer_cast<T>(ptr);
            }

            /// @brief 获取消息内容指针
            /// @tparam T 需要要解析的消息类型
            /// @return 消息内容指针
            /// @note 建议使用共享指针返回，避免大对象拷贝
            template <typename T>
            std::shared_ptr<T> cast_ptr() const
            {
                return std::static_pointer_cast<T>(ptr);
            }

            /// @brief 获取消息内容指针
            /// @tparam T 需要要解析的消息类型
            /// @return 消息内容指针
            /// @note 建议使用共享指针返回，避免大对象拷贝
            template <typename T>
            std::shared_ptr<T> cast_ptr()
            {
                return std::static_pointer_cast<T>(ptr);
            }

            /// @brief 判断是否有消息内容
            /// @return 是否有消息内容
            inline bool has_value() const
            {
                return static_cast<bool>(ptr);
            }
        } message_wrapper_type;

        // 回调函数类型，接受消息主题、消息内容
        typedef std::function<void(const topic_type &, message_wrapper_type)> callback_type;

        /// @brief 任务结构体 - 替代 std::tuple 提供更好的可读性
        struct task_entry_t
        {
            topic_type topic;                // 消息主题
            int32_t subscriber_id;           // 订阅者ID
            std::function<void()> task_func; // 任务函数

            // 构造函数
            task_entry_t() = default;
            task_entry_t(const topic_type &t, int32_t id, std::function<void()> func)
                : topic(t), subscriber_id(id), task_func(std::move(func)) {}

            // 移动构造函数
            task_entry_t(task_entry_t &&other) noexcept
                : topic(std::move(other.topic)),
                  subscriber_id(other.subscriber_id),
                  task_func(std::move(other.task_func)) {}

            // 移动赋值操作符
            task_entry_t &operator=(task_entry_t &&other) noexcept
            {
                if (this != &other)
                {
                    topic = std::move(other.topic);
                    subscriber_id = other.subscriber_id;
                    task_func = std::move(other.task_func);
                }
                return *this;
            }

            /// @brief 标记为取消（延迟删除优化）
            void mark_cancelled() noexcept
            {
                task_func = nullptr;
            }

            /// @brief 检查是否有效（未被取消且有任务函数）
            /// @return 是否有效，true：有效，false：无效
            inline bool is_valid() const noexcept
            {
                return static_cast<bool>(task_func);
            }

            // 禁用拷贝（避免std::function的拷贝开销）
            task_entry_t(const task_entry_t &) = delete;
            task_entry_t &operator=(const task_entry_t &) = delete;
        };

        /// @brief 构造函数
        /// @param auto_start 是否自动启动，默认为false
        /// @param task_stack_size 任务栈大小，默认为UTOOLS_COLLAB_TASK_STACK_DEFAULT_SIZE
        PubSub(bool auto_start = false, size_t task_stack_size = UTOOLS_COLLAB_TASK_STACK_DEFAULT_SIZE)
        {
            if (auto_start)
            {
                start(task_stack_size);
            }
        }

        virtual ~PubSub()
        {
            stop();
        }

        /// @brief 发布消息
        /// @tparam T 用户自定义消息类型
        /// @param topic 消息主题
        /// @param message 消息内容
        /// @note 完全依赖编译期类型推导，无运行时类型检查
        /// @warning 有序发布，如果有对应的相同类型的数据需要还没有处理完，会申请新的空间进行数据保存，并发布
        template <typename T>
        void publish(const topic_type &topic, T &&message)
        {
            ::utools::collab::LockGuard lock(__callback_mutex);
            auto it = __topics.find(topic);
            if (it == __topics.end())
            {
                // 初次创建主题 - 编译期类型推导
                auto &node = __topics[topic];
                node.set_message(std::forward<T>(message));
                return; // 初次创建，没有订阅者，直接返回
            }

            __publish<T>(topic, std::forward<T>(message), it);
        }

        /// @brief 刷新消息，标记指定主题的所有消息为取消
        /// @param topic 指定的主题
        /// @note 使用随机访问队列的手动锁定机制，避免重复加锁开销
        inline void flush(const topic_type &topic)
        {
            __tasks.for_each([topic](task_entry_t &task)
                             {
                    if (task.topic == topic)
                    {
                        task.mark_cancelled();
                    } });
        }

        /// @brief 发布消息，删除所有消息
        /// @param topic 消息主题
        /// @param message 消息内容
        /// @note 仅在启用消息刷新时有效
        template <typename T>
        void unique_publish(const topic_type &topic, T &&message)
        {
            flush(topic); // 删除指定主题的所有消息
            publish(topic, std::forward<T>(message));
        }

        /// @brief 替换消息，将已经在队列中的消息替换为新消息
        /// @tparam T 用户自定义消息类型
        /// @param topic 消息主题
        /// @param message 消息内容
        /// @note 仅在启用消息刷新时有效，如果队列中没有消息，则直接发布新消息
        template <typename T>
        void replace_publish(const topic_type &topic, T &&message)
        {
            ::utools::collab::LockGuard lock(__callback_mutex);
            auto it = __topics.find(topic);
            // 如果主题不存在，则初始化默认消息 - 编译期类型安全
            if (it == __topics.end())
            {
                auto &node = __topics[topic];
                node.set_message(std::forward<T>(message));
                return; // 初次创建，没有订阅者，直接返回
            }

            // 先更新last_msg到topic节点
            it->second.set_message(std::forward<T>(message));
            auto store_msg = it->second.last_msg;

            // 在队列中直接替换对应topic的任务
            int32_t replaced_count = 0;
            __tasks.for_each([&](task_entry_t &task)
                             {
                if (task.topic == topic)
                {
                    // 查找对应的订阅者ID是否存在
                    auto callbackIt = std::find_if(it->second.callbacks.begin(),
                                                   it->second.callbacks.end(),
                                                   [&task](const callback_entry_ptr_type &entry) {
                                                       return entry->id == task.subscriber_id;
                                                   });

                    if (callbackIt != it->second.callbacks.end())
                    {
                        // 直接替换任务函数，使用新的消息
                        task.task_func = [callbackPtr = callbackIt, topicCopy = topic, messageCopy = store_msg]() -> void {
                            (*callbackPtr)->callback(topicCopy, message_wrapper_type(messageCopy));
                        };
                        ++replaced_count;
                    }
                } });

            // 如果没有替换任何任务，说明队列中没有该topic的任务，直接发布新消息
            if (replaced_count == 0)
            {
                __publish<T>(topic, std::forward<T>(message), it);
            }
        }

        /// @brief 刷新消息，删除所有消息
        /// @note 仅在启用消息刷新时有效
        void flush()
        {
            __tasks.clear();
        }

        /// @brief 订阅消息
        /// @param topic 订阅的主题
        /// @param callback 回调函数
        /// @return 返回订阅ID
        /// @note 订阅ID可用于取消订阅
        virtual int32_t subscribe(const topic_type &topic, callback_type callback)
        {
            ::utools::collab::LockGuard lock(__callback_mutex);
            int32_t id = ++__current_id;
            __topics[topic].callbacks.emplace_back(std::make_shared<callback_entry_t>(id, callback));
            return id;
        }

        /// @brief 订阅消息
        /// @param topic 订阅的主题
        /// @param callback 回调函数
        /// @param default_message 默认初始化消息
        /// @return 返回订阅ID
        /// @note 订阅ID可用于取消订阅
        /// @note default_message 用于初始化消息，只有在第一次订阅时才会生效
        template <typename T>
        int32_t subscribe(const topic_type &topic, callback_type callback, T &&default_message)
        {
            ::utools::collab::LockGuard lock(__callback_mutex);
            decltype(__topics.find(topic)) it = __topics.end();
            it = __topics.find(topic);
            // 如果主题不存在，则初始化默认消息 - 编译期类型安全
            if (it == __topics.end())
            {
                auto &node = __topics[topic];
                node.set_message(std::forward<T>(default_message));
            }
            // 添加订阅者
            int32_t id = ++__current_id;
            __topics[topic].callbacks.emplace_back(std::make_shared<callback_entry_t>(id, callback));
            return id;
        }

        /// @brief 订阅消息
        /// @param topic 订阅的主题
        /// @param callback 回调函数
        /// @param obj 回调函数所属对象
        /// @return 返回订阅ID
        /// @note 订阅ID可用于取消订阅
        template <typename T>
        int32_t subscribe(const topic_type &topic,
                          void (T::*func)(const topic_type &, message_wrapper_type), T *obj)
        {
            auto callback = [obj, func](const topic_type &topic, message_wrapper_type message) -> void
            {
                (obj->*func)(topic, message);
            };

            return subscribe(topic, callback);
        }

        /// @brief 订阅消息
        /// @param topic 订阅的主题
        /// @param callback 回调函数
        /// @param obj 回调函数所属对象
        /// @param default_message 默认初始化消息
        /// @return 返回订阅ID
        /// @note 订阅ID可用于取消订阅
        /// @note default_message 用于初始化消息，只有在第一次订阅时才会生效
        template <typename T, typename U>
        int32_t subscribe(const topic_type &topic,
                          void (T::*func)(const topic_type &, message_wrapper_type), T *obj,
                          U &&default_message)
        {
            auto callback = [obj, func](const topic_type &topic, message_wrapper_type message) -> void
            {
                (obj->*func)(topic, message);
            };

            return subscribe(topic, callback, std::forward<U>(default_message));
        }

        /// @brief 取消订阅消息主题
        /// @param topic 消息主题
        /// @param callback_id 回调函数ID
        virtual void unsubscribe(const topic_type &topic, const int32_t callback_id)
        {
            ::utools::collab::LockGuard lock(__callback_mutex);
            auto it = __topics.find(topic);
            if (it != __topics.end())
            {
                auto &entries = it->second.callbacks;
                // 使用 erase-remove 惯用法
                entries.erase(
                    std::remove_if(entries.begin(), entries.end(),
                                   [callback_id](const callback_entry_ptr_type &entry)
                                   {
                                       return entry->id == callback_id;
                                   }),
                    entries.end());
            }
        }

        /// @brief 删除指定主题的所有订阅者
        /// @param topic 主题
        virtual void remove(const topic_type &topic)
        {
            ::utools::collab::LockGuard lock(__callback_mutex);
            __topics.erase(topic);
        }

        /// @brief 更新消息主题中的数据（纯编译期类型安全）
        /// @tparam T 消息类型（编译期确定）
        /// @param topic 主题
        /// @param message 消息
        /// @note 完全依赖编译期类型推导，无运行时类型检查
        template <typename T>
        void update(const topic_type &topic, T &&message)
        {
            ::utools::collab::LockGuard lock(__callback_mutex);
            auto it = __topics.find(topic);
            if (it != __topics.end())
            {
                it->second.set_message(std::forward<T>(message));
            }
            else
            {
                // 创建新节点并设置消息 - 编译期类型推导
                auto &node = __topics[topic];
                node.set_message(std::forward<T>(message));
            }
        }

        /// @brief 获取指定主题中的最新消息（纯编译期类型安全）
        /// @tparam T 期望的消息类型（编译期确定）
        /// @param topic 主题
        /// @return 类型安全的消息引用
        /// @note 调用者必须保证类型正确，编译期强制类型安全
        template <typename T>
        inline const T &get(const topic_type &topic) const
        {
            ::utools::collab::LockGuard lock(__callback_mutex);
            return __topics.at(topic).template get_message<T>();
        }

        /// @brief 获取指定主题中的最新消息（非const版本）
        /// @tparam T 期望的消息类型（编译期确定）
        /// @param topic 主题
        /// @return 类型安全的消息引用
        template <typename T>
        inline T &get(const topic_type &topic)
        {
            ::utools::collab::LockGuard lock(__callback_mutex);
            return __topics.at(topic).template get_message<T>();
        }

        /// @brief 获取指定主题中的最新消息（带默认值）
        /// @param topic 主题
        /// @param default_message 默认消息
        /// @return 类型安全的消息引用
        template <typename T>
        T &get(const topic_type &topic, T &&default_message)
        {
            {
                ::utools::collab::LockGuard lock(__callback_mutex);
                auto it = __topics.find(topic);
                if (it != __topics.end())
                {
                    return it->second.template get_message<T>();
                }
                // NOTE: 需要释放锁，否则会造成死锁
            }

            update(topic, std::forward<T>(default_message));
            return get<T>(topic);
        }

        /// @brief 判断缓存中是否存在指定主题的数据
        /// @param key 键
        /// @return bool 存在返回true，否则返回false
        bool contains(const topic_type &topic) const
        {
            ::utools::collab::LockGuard lock(__callback_mutex);
            return __topics.find(topic) != __topics.end();
        }

        /// @brief 启动订阅服务
        /// @param task_stack_size 任务栈大小
        void start(size_t task_stack_size = UTOOLS_COLLAB_TASK_STACK_DEFAULT_SIZE)
        {
            ::utools::collab::LockGuard lock(__callback_mutex);
            if (__running)
            {
                return;
            }
            __running = true;
            __work_task.bind(&PubSub::__process_task_queue, this).start(task_stack_size);
        }

        /// @brief 停止订阅服务
        void stop()
        {
            ::utools::collab::LockGuard lock(__callback_mutex);
            if (!__running)
            {
                return;
            }
            __running = false;

            auto stop_task = []()
            { return; };
            __tasks.emplace(topic_type{}, -1, std::move(stop_task));

            __tasks.notify_all();
            __work_task.stop();
        }

        /// @brief 判断订阅服务是否正在运行
        /// @return bool 返回true表示正在运行，否则返回false
        bool is_running() const
        {
            return __running;
        }

    protected:
        struct callback_entry_t
        {
            int32_t id;             // 订阅ID
            callback_type callback; // 回调函数

            callback_entry_t(int32_t id, callback_type callback) : id(id), callback(callback) {}
        };

        using callback_entry_ptr_type = std::shared_ptr<callback_entry_t>;

        /// @brief 消息节点
        struct _pubsub_node_t
        {
            message_type last_msg;                          // 最新消息
            std::vector<callback_entry_ptr_type> callbacks; // 回调函数列表

            template <typename T>
            inline void set_message(T &&message)
            {
                // 如果last_msg为空或类型不匹配，创建新的shared_ptr
                if (!last_msg || last_msg.use_count() > 1)
                {
                    last_msg = std::static_pointer_cast<void>(
                        std::make_shared<typename std::decay<T>::type>(std::forward<T>(message)));
                }
                else
                {
                    // 直接覆盖last_msg的内容，避免创建新对象
                    auto *ptr = static_cast<typename std::decay<T>::type *>(last_msg.get());
                    if (ptr)
                    {
                        *ptr = std::forward<T>(message);
                    }
                    else
                    {
                        // 类型不匹配，重新创建
                        last_msg = std::static_pointer_cast<void>(
                            std::make_shared<typename std::decay<T>::type>(std::forward<T>(message)));
                    }
                }
            }

            template <typename T>
            inline T &get_message()
            {
                return *std::static_pointer_cast<T>(last_msg);
            }
        };

        mutable ::utools::collab::Mutex __callback_mutex; // 保护任务队列的互斥量
        std::unordered_map<topic_type, _pubsub_node_t> __topics;

        // 自动修正BUFFER_SIZE为0时的默认值
        static constexpr std::size_t static_buffer_size = (BUFFER_SIZE == 0 ? 1024 : BUFFER_SIZE);
        ::utools::collab::SyncRandomAccessQueue<task_entry_t, static_buffer_size> __tasks;

        ::utools::collab::Task __work_task; // 工作线程任务

        int32_t __current_id{0}; // 订阅者ID
        bool __running{false};   // 是否运行

        typedef typename std::unordered_map<topic_type, struct _pubsub_node_t>::iterator topic_iterator_type;

        /// @brief 发布消息（MCU栈分配优化）
        /// @tparam T 用户自定义消息类型（编译期确定）
        /// @param topic 消息主题
        /// @param message 消息内容
        /// @param it 主题迭代器
        /// @note 支持栈/堆混合分配，保持回调兼容性
        template <typename T, bool is_replace = false>
        inline void __publish(const topic_type &topic, T &&message, topic_iterator_type &it)
        {
            it->second.set_message(std::forward<T>(message));
            auto store_msg = it->second.last_msg;

            // 批量优化：根据回调数量选择最优处理方式
            const auto &callbacks = it->second.callbacks;
            const auto callback_size = callbacks.size();

            // 单个回调，直接添加任务
            if (callback_size == 1)
            {
                const auto entry = callbacks.front();
                auto task = [entry, topic, store_msg]() -> void
                {
                    entry->callback(topic, message_wrapper_type(store_msg));
                };
                __tasks.emplace(topic, entry->id, std::move(task));
            }
            else if (callback_size > 1)
            {
                // 批量添加，逐个添加任务到随机访问队列，减少锁竞争
                auto func = [topic, store_msg](const callback_entry_ptr_type &entry)
                {
                    auto task = [entry, topic, store_msg]()
                    {
                        entry->callback(topic, message_wrapper_type(store_msg));
                    };
                    return task_entry_t(topic, entry->id, std::move(task));
                };

                __tasks.push_batch(callbacks.begin(), callbacks.end(), func);
            }
        }

        void __process_task_queue()
        {
            while (__running)
            {
                auto task = __tasks.pop_wait();

                // 延迟删除优化：检查任务是否有效
                if (task.is_valid())
                {
                    // 提取任务函数
                    auto task_func = std::move(task.task_func);

                    task_func();
                }
                // 如果任务无效（被标记删除），则跳过执行
            }
        }
    };
#endif // UTOOLS_PUBSUB_ENABLE && UTOOLS_COLLAB_TASK_ENABLE

} // namespace utools::pattern


#endif // __UTOOLS_PUBSUB_H__
